package cn.kgc.listener;

import cn.kgc.entity.Messageinfo;
import cn.kgc.service.ConsumerMessageInfoService;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Value;

import javax.annotation.Resource;

/**
 * @author wangyue
 * @version 1.0
 */
@SuppressWarnings("all")
public abstract class AbstractMessageListener implements ChannelAwareMessageListener{
    @Resource
    private ConsumerMessageInfoService consumerMessageInfoService;

    @Value("${mq.maxRetryCount}")
    private int maxRetryCount;

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        //1 判断消息是否是第一次被消费,如果是第一次被消费向数据库中插入数据。否则更新消息的消费次数。
        //获得路由的信息
       String exchange= message.getMessageProperties().getReceivedExchange();
       //获得消息的id
       String messageId= message.getMessageProperties().getHeader("spring_returned_message_correlation");
       //路由key
       String receivedRoutingKey = message.getMessageProperties().getReceivedRoutingKey();
       //获得消息内容
       String jsonStr=new String(message.getBody());
       //获得接收队列名称
       String consumerQueue = message.getMessageProperties().getConsumerQueue();
       //获得消息的延迟时间
       int delay=(Integer)message.getMessageProperties().getHeaders().get("delay");
       //获得消息
        Messageinfo messageinfo=consumerMessageInfoService.messageReceiveReady(
                    messageId,exchange, receivedRoutingKey,consumerQueue,jsonStr,delay
        );
        long tag=message.getMessageProperties().getDeliveryTag();
        //2 判断消息的消费次数是否超过最多次数,如果消息的消费次数超过最大次数,将消息保存到死信队列中。
        if(messageinfo.getSequence()>maxRetryCount){
            /*
             * tag: 消息在通道中的编号
             * multiple: 是否批量处理
             * requeue: 是否重新入队
             */
            channel.basicNack(tag,false,false);
            return;
        }
        try {
            //3 如果消息的消费次数没有超过最大次数,消费者可以处理此消息。
            receiveMessage(message,channel);
            //消费者确认处理消息
            channel.basicAck(tag,true);
            //数据库中删除消息
            consumerMessageInfoService.messageReceiveSuccess(messageId);
        } catch (Exception e) {
            //4 如果处理消息时产生了异常(没有正确处理消息)将消息重新返回到列中,消费者下次进行消费。
            Thread.sleep(500L * messageinfo.getSequence());
            channel.basicNack(tag,false,true);
        }
    }

    /**
     * 业务需要真正处理的方法
     */
    public abstract void receiveMessage(Message message, Channel channel) throws Exception;
}
